package spring.cloud.tasks.tasks_manager.tasks_manager.service;

import com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.KeeperException;
import spring.cloud.tasks.tasks_manager.tasks_manager.listener.TaskConfigListener;
import spring.cloud.tasks.tasks_manager.tasks_manager.listener.TaskExecutorOnlineOfflineListener;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;

import java.util.List;

/**
 * 配置监听和执行器上下线
 */
@Slf4j
@Setter
@Getter
public class TaskListenersService {
    private String namespace;
    private CuratorFramework curatorFramework;
    private TreeCacheService treeCacheService;
    //
    private AllotTaskItemService allotTaskItemService;

    public TaskListenersService(String namespace,//
                                CuratorFramework curatorFramework,//
                                TreeCacheService treeCacheService, //
                                AllotTaskItemService allotTaskItemService//
    ) {
        this.namespace = namespace;
        this.curatorFramework = curatorFramework;
        this.treeCacheService = treeCacheService;
        this.allotTaskItemService = allotTaskItemService;
    }

    public void addExistTaskListener() throws Exception {
        if (null != curatorFramework.checkExists().forPath(Node.TASKS_NODE_PATH)) {
            List<String> taskIdList = curatorFramework.getChildren().forPath(Node.TASKS_NODE_PATH);
            log.info("addExistJobPathListener, jobs = {}", taskIdList);
            if (taskIdList != null) {
                for (String taskId : taskIdList) {
                    addTaskListener(taskId);
                }
            }
        }
    }

    public void addTaskListener(String taskId) throws Exception {
        if (addTaskConfigListener(taskId)) {
            addTaskExecutorListener(taskId);
        }
    }

    /**
     * Add job config path listener
     *
     * @return false, if the job/config path is not existing
     */
    private boolean addTaskConfigListener(String taskId) throws Exception {
        String path = Node.TASKS_NODE_PATH + "/" + taskId + "/config";
        int waitConfigPathCreatedCounts = 50;
        do {
            waitConfigPathCreatedCounts--;
            if (curatorFramework.checkExists().forPath(path) != null) {
                break;
            }
            if (waitConfigPathCreatedCounts == 0) {
                String fullPath = namespace + path;
                log.warn("the path {} is not existing", fullPath);
                return false;
            }
            Thread.sleep(1000L);
        } while (true);

        int depth = 0;
        List<String> list = Lists.newArrayList(Node.getJobConfigEnableNodePath(taskId), Node.getTaskConfigForceShardNodePath(taskId));
        for (String $path : list) {
            treeCacheService.addTreeCacheIfAbsent($path, depth);
            treeCacheService.addTreeCacheListenerIfAbsent($path, depth, new TaskConfigListener(taskId, allotTaskItemService));
        }

        return true;
    }

    private void addTaskExecutorListener(String taskId) throws Exception {
        String path = Node.getTaskExecutorsNodePath(taskId);
        if (curatorFramework.checkExists().forPath(path) == null) {
            try {
                curatorFramework.create().creatingParentsIfNeeded().forPath(path);
            } catch (KeeperException.NodeExistsException e) {// NOSONAR
                log.info("node {} already existed, so skip creation", path);
            }
        }
        int depth = 2;
        treeCacheService.addTreeCacheIfAbsent(path, depth);
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new TaskExecutorOnlineOfflineListener(taskId, allotTaskItemService));
    }

    //
    public void removeTaskTreeCache(String jobName) throws Exception {
        removeTaskConfigTreeCache(jobName);
        removeTaskExecutorTreeCache(jobName);
    }

    private void removeTaskConfigTreeCache(String taskId) throws Exception {
        int depth = 0;
        List<String> list = Lists.newArrayList(Node.getJobConfigEnableNodePath(taskId), Node.getTaskConfigForceShardNodePath(taskId));
        for (String path : list) {
            treeCacheService.removeTreeCache(path, depth);
        }
    }

    private void removeTaskExecutorTreeCache(String taskId) throws Exception {
        String path = Node.getTaskExecutorsNodePath(taskId);
        int depth = 2;
        treeCacheService.removeTreeCache(path, depth);
    }


}
